package cn.simon.free.redis.mq.core.producer;

import cn.simon.free.redis.mq.component.IdGeneratorSupport;
import cn.simon.free.redis.mq.config.properties.RedisMessageProperties;
import cn.simon.free.redis.mq.core.consumer.DelayRedisMessageConsumer;
import cn.simon.free.redis.mq.core.consumer.NormalRedisMessageConsumer;
import cn.simon.free.redis.mq.exceptions.MessageTypeException;
import cn.simon.free.redis.mq.model.Message;
import cn.simon.free.redis.mq.model.delay.DelayMessage;
import cn.simon.free.redis.mq.model.normal.NormalMessage;
import lombok.AllArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @author simon
 * @date 2021/7/15 15:32
 * @copyright:
 * @modifier
 * @motto a no talent, people who have to study hard
 */
@Component
@AllArgsConstructor(onConstructor = @__(@Autowired))
public class RedisMessageProducer implements Producer<Message> {

    private final NormalRedisMessageProducer normalRedisMessageProducer;
    private final DelayRedisMessageProducer delayRedisMessageProducer;
    private final RedisMessageProperties redisMessageProperties;
    private final IdGeneratorSupport idGeneratorSupport;

    /**
     * 发送普通消息(任务id自动生成)
     *
     * @param queueName
     * @param message
     * @return
     */
    public String send(String queueName, Object message) {
        return send(queueName, idGeneratorSupport.nextId(), message);
    }

    /**
     * 发送普通消息
     *
     * @param queueName
     * @param taskId
     * @param message
     * @return
     */
    public String send(String queueName, Long taskId, Object message) {
        return selector(queueName).send(queueName, NormalMessage.builder()
                .taskId(taskId)
                .message(message)
                .build());
    }

    /**
     * 发送延迟消息(任务id自动生成)
     *
     * @param queueName
     * @param message
     * @param delayTime
     * @param timeUnit
     * @return
     */
    public String send(String queueName, Object message, long delayTime, TimeUnit timeUnit) {
        return send(queueName, message, idGeneratorSupport.nextId(), delayTime, timeUnit);
    }

    /**
     * 发送延迟消息
     *
     * @param queueName
     * @param message
     * @param taskId
     * @param delayTime
     * @param timeUnit
     * @return
     */
    public String send(String queueName, Object message, Long taskId, long delayTime, TimeUnit timeUnit) {
        Message build = DelayMessage.builder()
                .taskId(taskId)
                .message(message)
                .delayTime(System.currentTimeMillis() + timeUnit.toMillis(delayTime))
                .build();
        return selector(queueName).send(queueName, build, delayTime, timeUnit);
    }

    @Override
    public String send(String queueName, Message message) {
        return selector(queueName).send(queueName, message);
    }

    @Override
    public String send(String queueName, Message message, long delayTime) {
        return selector(queueName).send(queueName, message, delayTime);
    }

    @Override
    public String send(String queueName, Message message, long delayTime, TimeUnit timeUnit) {
        return selector(queueName).send(queueName, message, delayTime, timeUnit);
    }

    private Producer selector(String queueName) {
        Assert.notNull(queueName, "不能为空");
        //实例bean
        Object redisMessageConsumer =
                Optional.ofNullable(redisMessageProperties.getQueueNameWithBeanHash().getOrDefault(queueName, null)).orElseThrow(() -> new MessageTypeException(String.format("队列:%s,不存在,请检查..", queueName)));
        if (NormalRedisMessageConsumer.class.isAssignableFrom(redisMessageConsumer.getClass())) {
            return normalRedisMessageProducer;
        } else if (DelayRedisMessageConsumer.class.isAssignableFrom(redisMessageConsumer.getClass())) {
            return delayRedisMessageProducer;
        } else {
            throw new MessageTypeException(String.format("队列:%s,不支持的消息类型", queueName));
        }
    }

}
